package com.hanxiaozhang.example.listener.mgsconsumer;

import com.hanxiaozhang.constant.RocketConstant;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.stereotype.Component;

import java.io.UnsupportedEncodingException;
import java.util.List;

/**
 * 〈一句话功能简述〉<br>
 * 〈Subscribe模式pull〉
 * <p>
 * 在subscribe模式下，同一个消费组下的多个LitePullConsumer会负载均衡消费，与PushConsumer一致
 *
 * @author hanxinghua
 * @create 2022/10/3
 * @since 1.0.0
 */
@Slf4j
@Component
public class No7LitePullSubscribeMsgOriginalSyntax {


    /**
     * 拉去消息
     *
     * @param pullBatchSize
     * @throws Exception
     */
    public void pull(int pullBatchSize) {
        Long startTime = System.currentTimeMillis();
        DefaultLitePullConsumer litePullConsumer = null;
        try {
            // 1. 初始化DefaultLitePullConsumer，并设置ConsumerGroupName
            litePullConsumer = new DefaultLitePullConsumer(RocketConstant.LITE_PULL_SUBSCRIBE_CONSUMER_GROUP);
            // 2. 调用subscribe方法订阅topic，并启动
            litePullConsumer.subscribe(RocketConstant.LITE_PULL_SUBSCRIBE_TOPIC, "*");
            // 设置每一次拉取的最大消息数量
            litePullConsumer.setPullBatchSize(pullBatchSize);
            litePullConsumer.start();
            // 轮询 10s
            while ((System.currentTimeMillis() - startTime) < 10 * 1000) {
                log.info("循环");
                // LitePullConsumer拉取消息调用的是轮询poll接口，如果能拉取到消息则返回对应的消息列表，否则返回null。
                // LitePullConsumer默认是自动提交位点
                List<MessageExt> messageExts = litePullConsumer.poll();
                messageExts.forEach(x->{
                    try {
                        log.info("{}收到消息：{}", this.getClass().getSimpleName(),  new String(x.getBody(), "utf-8"));
                    } catch (UnsupportedEncodingException e) {
                        e.printStackTrace();
                    }
                });
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (litePullConsumer != null) {
                litePullConsumer.shutdown();
            }
        }
    }

}
